package com.com.zs21cp.wordcountDemo;

import com.com.zs21cp.ToolBean.ClickSource;
import com.com.zs21cp.ToolBean.WaterSensor;
import com.com.zs21cp.ToolBean.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class TopNprocessDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        DataStreamSource<String> hadoop2 = env.socketTextStream("hadoop2", 7777);
        SingleOutputStreamOperator<WaterSensor> map = hadoop2.map(new WaterSensorMapFunction());

        //流处理分配watermark必须在map算子之后，将流转化后进行分配，如果返回的是DataStreamSource，则需要算子进
        // 行进一步的转化，如果是DataSource,则不需要算子进行转化，因为一个是流处理一个是批处理，不知道为什么流处理不能直
        // 接进行watermark的分配
        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator =
            map.assignTimestampsAndWatermarks(
            WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(10)).
            withTimestampAssigner(((element, ts) -> element.getTs() * 1000L)));


        env.execute();

    }
}
